package com.education.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.app.BaseAppV1;
import com.education.realtime.bean.AnswerRightUVBean;
import com.education.realtime.bean.PaperDistributionBean;
import com.education.realtime.common.Constant;
import com.education.realtime.util.AtguiguUtil;
import com.education.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsPaperDistribution extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsPaperDistribution().init(3153, "DwsPaperDistribution", Constant.TOPIC_DWD_EXAM);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 解析数据, 过滤出各个分数段
        SingleOutputStreamOperator<PaperDistributionBean> beanStream = findDistribution(stream);
        // 2. 开窗聚合
        SingleOutputStreamOperator<PaperDistributionBean> aggregatedStream = windowAndAggregate(beanStream);
        aggregatedStream.print();

//        // 3. 数据写出到clickhouse中
//        writeToClickHouse(aggregatedStream);

    }

//    private void writeToClickHouse(SingleOutputStreamOperator<PaperDistributionBean> aggregatedStream) {
//        aggregatedStream.addSink(FlinkSinkUtil.getClickHouse(Constant.CLICKHOUSE_DB, "dws_exam_paper_distribution_window", PaperDistributionBean.class));
//    }

    private SingleOutputStreamOperator<PaperDistributionBean> windowAndAggregate(SingleOutputStreamOperator<PaperDistributionBean> beanStream) {
       return beanStream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<PaperDistributionBean>() {
                            @Override
                            public PaperDistributionBean reduce(PaperDistributionBean b1,
                                                                PaperDistributionBean b2) throws Exception {
                                b1.setA(b1.getA() + b2.getA());
                                b1.setB(b1.getB() + b2.getB());
                                b1.setC(b1.getC() + b2.getC());
                                b1.setD(b1.getD() + b2.getD());
                                b1.setE(b1.getE() + b2.getE());
                                return b1;
                            }
                        }, new ProcessAllWindowFunction<PaperDistributionBean, PaperDistributionBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx,
                                                Iterable<PaperDistributionBean> elements,
                                                Collector<PaperDistributionBean> out) throws Exception {
                                String stt = AtguiguUtil.toDateTime(ctx.window().getStart());
                                String edt = AtguiguUtil.toDateTime(ctx.window().getEnd());
                                PaperDistributionBean bean = elements.iterator().next();
                                bean.setStt(stt);
                                bean.setEdt(edt);
                                bean.setTs(System.currentTimeMillis());
                                out.collect(bean);
                            }
                        }

                );
    }

    private SingleOutputStreamOperator<PaperDistributionBean> findDistribution(DataStreamSource<String> stream) {
        return  stream.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((obj,ts)->obj.getLong("ts")*1000)
                )
                .keyBy(obj->obj.getString("paper_id"))
                  .process(new KeyedProcessFunction<String, JSONObject, PaperDistributionBean>() {

                      private ValueState<Long> a;
                      private ValueState<Long> b;
                      private ValueState<Long> c;
                      private ValueState<Long> d;
                      private ValueState<Long> e;

                      @Override
                      public void open(Configuration parameters) throws Exception {
                          a = getRuntimeContext().getState(new ValueStateDescriptor<Long>("a", Long.class));
                          b = getRuntimeContext().getState(new ValueStateDescriptor<Long>("b", Long.class));
                          c = getRuntimeContext().getState(new ValueStateDescriptor<Long>("c", Long.class));
                          d = getRuntimeContext().getState(new ValueStateDescriptor<Long>("d", Long.class));
                          e = getRuntimeContext().getState(new ValueStateDescriptor<Long>("e", Long.class));
                      }

                      @Override
                      public void processElement(JSONObject obj,
                                                 Context ctx,
                                                 Collector<PaperDistributionBean> out) throws Exception {
                          long UV=0L;
                          Long ts = obj.getLong("ts");
                          String today = AtguiguUtil.toDate(ts);
                          Integer score = obj.getInteger("score");
                          if (score<60){
                              UV+=1;
                              a.update(UV);
                          }else if (score>=60 && score<=70 ){
                              UV+=1;
                              b.update(UV);
                          }
                          else if (score>70 && score<=80 ){
                              UV+=1;
                              c.update(UV);
                          }else if (score>80 && score<=90 ){
                              UV+=1;
                              d.update(UV);
                          }else {
                              UV+=1;
                              e.update(UV);
                          }
                          PaperDistributionBean bean = new PaperDistributionBean("",
                                  "",
                                  a.value(),
                                  b.value(),
                                  c.value(),
                                  d.value(),
                                  e.value(),
                                  ts);
                          out.collect(bean);
                      }
                  });
    }
}
